package cn.edu.laud.sany.datatest;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

import cn.edu.laud.sany.datatest.RandomKeys.Iterm;
import cn.edu.laud.sany.utils.TimeUtils;

public class ThreadWorker extends Thread {

	public String host;
	public Integer port;
	private String url;
	public Integer no;
	public int type=0;//0 count 1 read


	public ThreadWorker(Integer no,String host, Integer port,int type) {
		super();
		this.no=no;
		this.host = host;
		this.port = port;
		this.type=type;
		this.url = "jdbc:cassandra://"+host+":"+port+"?version=2.0.0";
	}
	@Override
	public void run(){
		long result=0;
		if(type==0){
			try {
				result=getCount();
			} catch (IOException e) {
				e.printStackTrace();
			} catch (InvalidRequestException e) {
				e.printStackTrace();
			} catch (UnavailableException e) {
				e.printStackTrace();
			} catch (TimedOutException e) {
				e.printStackTrace();
			} catch (TException e) {
				e.printStackTrace();
			}
		}else{
			try {
				result=bigTableTest();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		Master.commitResult(no, result);
	}
	public long getCount() throws IOException, InvalidRequestException, UnavailableException, TimedOutException, TException{
		File file=null;

		TTransport tr = new TFramedTransport(new TSocket(host, port));
		TProtocol proto = new TBinaryProtocol(tr);

		Cassandra.Client client = new Cassandra.Client(proto);
		tr.open();
		client.set_keyspace(Master.keyspace);
		long sum=0;
		String maxCf="";
		long max=0;
		while((file=Master.askForACountJob(no))!=null){
			System.out.println(file.getName());
			BufferedReader bReader=new BufferedReader(new FileReader(file));
			String string=null;
			ColumnParent column_parent=new ColumnParent(file.getName());
			SlicePredicate slicePredicate=new SlicePredicate();
			SliceRange sliceRange =new SliceRange(ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false,Integer.MAX_VALUE);
			slicePredicate.setSlice_range(sliceRange);
			int count=0;
			while((string=bReader.readLine())!=null){
				count=client.get_count(ByteBufferUtil.bytes(string), column_parent, slicePredicate, ConsistencyLevel.ALL);
				sum+=count;
				if(count>max){
					max=count;
					maxCf=file.getName();
				}
			}
		}
		System.out.println("maxCf:"+maxCf+",max size:"+max);
		tr.close();
		return sum;
	}
	
	
	static{
		try {
			Class.forName("org.apache.cassandra.cql.jdbc.CassandraDriver");
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
	}
	public long bigTableTest() throws SQLException{
		Connection laudConnection=null;
		Statement laudStatement=null;
		try {
			laudConnection = DriverManager.getConnection(url, "", "");
			laudStatement = laudConnection.createStatement();
			laudStatement.execute("use "+Master.keyspace);
		} catch (SQLException e1) {
			e1.printStackTrace();
		}
		long totalcost=0;
		List<Iterm> list=Master.askForAReadJob(no);
		System.out.println("begin to query...");
		for(Iterm iterm:list){
			long start=System.nanoTime();
			laudStatement.execute("select * from  "+iterm.cf+" where equip='"+iterm.key+"'");
			ResultSet rSet=laudStatement.getResultSet();
//			while (rSet.next()) {
//				for(int i=1;i<=rSet.getMetaData().getColumnCount();i++){
//					Object object=	rSet.getObject(i);
//					System.out.println(object);
//				}
//			}
			long cost=System.nanoTime()-start;
			totalcost+=cost;
		}
		System.out.println(TimeUtils.nanoSecondToString(totalcost));
		laudStatement.close();
		laudConnection.close();
		return totalcost;
	}

}
